Pipelining & Shuffles

The speed of a Spark job is determined by how efficiently it processes data in memory (pipelining) versus how much it has to transfer data across the network (shuffling). To design high-performance applications, you must master the difference between Narrow Dependencies (Pipelining) and Wide Dependencies (Shuffling).

Pipelining (Narrow Dependencies)

When consecutive transformations in your code are Narrow Dependencies (meaning each partition of the parent DataFrame is used by at most one partition of the child DataFrame, like map or filter):

• Pipelining: Spark merges these transformations together into a single Stage.
• Instead of writing intermediate results to memory or disk, Spark processes data in a stream of records inside a single executor thread's RAM.
• Performance Cost: Extremely Low. Data is kept local to the CPU cache and RAM, resulting in optimal speed.

Shuffling (Wide Dependencies)

When a transformation is a Wide Dependency (meaning multiple child partitions depend on data from a single parent partition, like groupBy, distinct, join, or repartition):

• Shuffling: Spark is forced to split the job into a new Stage.
• The Shuffle Phase requires reorganizing and moving data across the cluster network so that rows sharing the same key land on the same worker node.

[Stage 1 Executors]  Write Shuffle Output to Local Disk

                    Network Data Transfer

[Stage 2 Executors]  Read Shuffle Input from Network

The 4 High-Overhead Phases of a Shuffle:

1. Disk Spilling (Map Side): Spark partitions the data based on key hashes and writes these temporary "shuffle blocks" to the executor's local disk.
2. Network Transfer: The target executors download their corresponding shuffle files from the parent executors across the network.
3. Deserialization / Serialization: Converting Java/Python objects to bytes to send across the network, and back into JVM objects on arrival.
4. Disk Reads (Reduce Side): The target executors read the downloaded blocks, sorting and merging the data in memory before executing the next stage's transformations.

Configuring Shuffle Partitions in PySpark

By default, whenever a shuffle occurs, Spark creates 200 shuffle partitions (which can result in very small partitions and high task scheduling overhead on small datasets, or too few partitions on huge datasets).

You can configure this using spark.sql.shuffle.partitions:

from pyspark.sql import SparkSession

# 1. Setup Spark
spark = SparkSession.builder \
    .appName("Pipelining and Shuffles") \
    .config("spark.sql.shuffle.partitions", "8") \
    .master("local[*]") \
    .getOrCreate()

# We set spark.sql.shuffle.partitions to 8 instead of the default 200.
# This prevents launching 200 separate parallel task threads on a small dataset!

# 2. Sample Data
data = [("Admin", 5000), ("Sales", 6000), ("Admin", 4000), ("Sales", 3000)]
df = spark.createDataFrame(data, ["dept", "salary"])

# 3. Trigger a Shuffle (groupBy)
shuffled_df = df.groupBy("dept").sum("salary")

# 4. Action (Triggers execution)
shuffled_df.show()

# If you check the Spark Web UI or execution metrics, you will see exactly 8 Tasks 
# were scheduled for the second stage, matching our shuffle partitions configuration!